# 导入 py4j 库，用于与 Java 程序通信（RocketMQ 是 Java 编写的）
from py4j.java_gateway import JavaGateway, GatewayParameters
import threading
import time
from app.core.config import settings

class RocketMQClient:
    # 初始化 RocketMQ 客户端，配置连接参数
    def __init__(self, namesrv_addr, producer_group, gateway_host="127.0.0.1", gateway_port=8084):
        # 创建 Java 网关连接，通过 py4j 与 Java 程序通信
        self.gateway = JavaGateway(
            # 配置网关参数
            gateway_parameters=GatewayParameters(
                # Java 网关地址
                address=gateway_host,
                # Java 网关端口
                port=gateway_port,
                # 自动类型转换
                auto_convert=True,
            )
        )
        # 获取 Java 程序入口点对象，用于调用 Java 方法
        self.proxy = self.gateway.entry_point
        # 设置 Name Server 地址，Name Server 是 RocketMQ 的服务发现组件
        self.namesrv_addr = namesrv_addr
        # 设置生产者组名，用于标识一组生产者
        self.producer_group = producer_group

    def start_producer(self):
        """启动生产者"""
        try:
            # 调用 Java 方法启动生产者，传入 Name Server 地址和生产者组名
            self.proxy.startProducer(self.namesrv_addr, self.producer_group)
            print("✅ 生产者启动成功")
            return True
        except Exception as e:
            print(f"❌ 启动生产者失败: {str(e)}")
            return False

    def send_message(self, topic, body, tags="default"):
        """发送消息"""
        try:            # 调用 Java 方法发送消息，传入主题(topic)、标签(tags)和消息体(body)
            result = self.proxy.sendMessage(topic, tags, body)
            print(f"✅ 消息发送成功: MsgID={result.getMsgId()}")
            return True
        except Exception as e:
            print(f"❌ 消息发送失败: {str(e)}")
            return False

    def shutdown(self):
        """关闭连接"""
        self.proxy.shutdownProducer() # 调用 Java 方法关闭生产者
        self.gateway.shutdown()  # 关闭 Java 网关连接
        print("✅ 生产者连接已关闭")


# 配置信息
NAMESRV_ADDR = "127.0.0.1:9876"  # 替换为实际地址
PRODUCER_GROUP = "python_producer_group"
TOPIC_NAME = "userMessages" #配置信息：主题名称

# 运行生产者的主函数
def run_producer():
    #  # 创建 RocketMQ 客户端实例
    mq_client = RocketMQClient(NAMESRV_ADDR, PRODUCER_GROUP)
    # 尝试启动生产者，如果启动失败则直接返回
    if not mq_client.start_producer():
        return

    print("\n⚡ 生产者已就绪，输入消息内容发送到MQ (输入'exit'退出):")
    try:
        # 循环接收用户输入并发送消息
        while True:
            # 获取用户输入的消息内容
            message = input("> ")
            # 如果输入 "exit" 则退出循环
            if message.lower() == "exit":
                print("接收到退出指令...")
                break
            if message.strip() != "": # 如果输入的不是空消息，则发送消息
                mq_client.send_message(TOPIC_NAME, message)
    finally:# 确保程序退出前关闭生产者连接
        mq_client.proxy.shutdownProducer()
        print("生产者程序已退出")


# ... existing code ...

class RocketMQManager:
    """RocketMQ 客户端管理器，使用单例模式管理连接"""
    _instance = None
    _initialized = False

    def __new__(cls): #new在init之前调用，负责实际创建并返回类的实例
        if cls._instance is None: #首次调用时，cls._instance为None，创建新实例并保存
            cls._instance = super(RocketMQManager, cls).__new__(cls)
        return cls._instance

    def __init__(self):
        if not self._initialized:


            # 从配置中获取 RocketMQ 设置
            self.namesrv_addr = getattr(settings, 'ROCKETMQ_NAMESRV_ADDR', '127.0.0.1:9876')
            self.producer_group = getattr(settings, 'ROCKETMQ_PRODUCER_GROUP', 'python_producer_group')
            self.topic_name = getattr(settings, 'ROCKETMQ_TOPIC_NAME', 'userMessages')

            # 初始化客户端
            self.mq_client = RocketMQClient(
                self.namesrv_addr,
                self.producer_group,
                gateway_host=getattr(settings, 'ROCKETMQ_GATEWAY_HOST', '127.0.0.1'),
                gateway_port=getattr(settings, 'ROCKETMQ_GATEWAY_PORT', 8084)
            )

            self.started = False
            self._initialized = True

    def start_producer(self):
        """启动生产者（如果尚未启动）"""
        if not self.started:
            self.started = self.mq_client.start_producer()
        return self.started

    def send_message(self, body, tags="default", topic=None):
        """发送消息"""
        if not self.started:
            if not self.start_producer():
                return False

        topic_to_use = topic or self.topic_name
        return self.mq_client.send_message(topic_to_use, body, tags)

    def shutdown(self):
        """关闭生产者"""
        if self.started:
            self.mq_client.shutdown()
            self.started = False


# 全局实例
rocketmq_manager = RocketMQManager()

if __name__ == "__main__":
    run_producer()